package com.tjhk.common.utils.thread.async;

import com.tjhk.common.utils.thread.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;

/**
 * @author ：little grape
 * @description：TODO
 * @date ：2023/7/16 18:35
 * 批处理任务处理工具，可实现线程自动扩容、缩容
 * maxRecordSize 最大批量处理记录数
 * keepAliveTime ThreadPoolExecutor keepAliveTime
 * RejectPolicy 线程异常策略
 * taskMethod 批处理任务方法
 * 示例代码
//        BathConsumeUtil<String> bathConsumeUtil = new BathConsumeUtil<>(4,8,100,"test",1000L);
//        注册批处理方法
//          bathConsumeUtil.registerMethod(list -> {
//              System.out.println("批处理方法将处理记录数："+list.size()+"=="+ String.join(",", list));
//              ThreadUtil.sleep(1000);
//          });
//        启动
//          bathConsumeUtil.start();
//          new Thread(()->{
//              for (int i = 0; i < 50000; i++) {
//                  bathConsumeUtil.add("结果"+i);
//                  ThreadUtil.sleep(1);
//              }
//              for (int i = 0; i < 30000; i++) {
//                  bathConsumeUtil.add("结果"+i);
//                  ThreadUtil.sleep(new Random(10).nextInt(10));
//              }
//          }).start();
 *
 *
 */
public class BathConsumeUtil<T> {
//    public static void main(String[] args) {
//         BathConsumeUtil<String> bathConsumeUtil = new BathConsumeUtil<>(4,8,100,"test",1000L);
//          bathConsumeUtil.registerMethod(list -> {
//              System.out.println("批处理方法将处理记录数："+list.size()+"=="+ String.join(",", list));
//              ThreadUtil.sleep(1000);
//          });
//          bathConsumeUtil.start();
//          new Thread(()->{
//              for (int i = 0; i < 10000; i++) {
//                  bathConsumeUtil.add("结果"+i);
//                  ThreadUtil.sleep(new Random(10).nextInt(100));
//              }
//              System.out.println("====================================================");
//              for (int i = 0; i < 50000; i++) {
//                  bathConsumeUtil.add("结果"+i);
//                  ThreadUtil.sleep(1);
//              }
//          }).start();
//    }
    private static final Logger log = LoggerFactory.getLogger(BathConsumeUtil.class);

    private volatile boolean stopped = false;
    private final int maxRecordSize;
    private int keepAliveTime = 1;
    private TimeUnit timeUnit = TimeUnit.SECONDS;
    private final ThreadPoolExecutor executor;
    private final int poolSize;
    private final int maximumPoolSize;
    private RejectPolicy rejectPolicy = RejectPolicy.DISCARD;
    private final ConcurrentLinkedQueue<T> queue = new ConcurrentLinkedQueue<>();
    private Consumer<List<T>> taskMethod;
    private String taskName="task";

    /**等待任务时间**/
    private long maxWaitTime = 0L;
    private final int minCorePoolSize = 2;
    private final AtomicBoolean flag = new AtomicBoolean(false);
    private boolean enableTimer = false;
    private boolean toAdd = false;
    private boolean toRemove = false;

    public BathConsumeUtil(int maxRecordSize) {
        this.maxRecordSize = maxRecordSize;
        this.poolSize = Runtime.getRuntime().availableProcessors();
        this.maximumPoolSize = this.poolSize * 2;
        this.executor = new ThreadPoolExtUtil(this.poolSize,this.maximumPoolSize,this.keepAliveTime,this.timeUnit,
                new LinkedBlockingQueue<>(64),
                new NamedThreadFactory("BathConsume-"+ this.taskName+"-",true),
                new RejectedPolicyWithReport(this.rejectPolicy));
    }

    public BathConsumeUtil(int poolSize, int maximumPoolSize, int maxRecordSize, String taskName) {
        this.maxRecordSize = maxRecordSize;
        this.poolSize = poolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.taskName = taskName;
        this.executor = new ThreadPoolExtUtil(this.poolSize,this.maximumPoolSize,this.keepAliveTime,this.timeUnit,
                new LinkedBlockingQueue<>(64),
                new NamedThreadFactory("BathConsume-"+ this.taskName+"-",true),
                new RejectedPolicyWithReport(this.rejectPolicy));
    }
    public BathConsumeUtil(int poolSize, int maximumPoolSize, int maxRecordSize, String taskName, long maxWaitTime) {
        this(poolSize,
                maximumPoolSize,
                maxRecordSize,
                taskName);
        this.maxWaitTime = maxWaitTime;
    }
    public BathConsumeUtil(int poolSize, int maximumPoolSize, int maxRecordSize, String taskName, long maxWaitTime, int keepAliveTime, TimeUnit timeUnit) {
        this.maxRecordSize = maxRecordSize;
        this.keepAliveTime = keepAliveTime;
        this.poolSize = poolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.taskName = taskName;
        this.timeUnit = timeUnit;
        this.maxWaitTime = maxWaitTime;
        this.executor = new ThreadPoolExtUtil(this.poolSize,this.maximumPoolSize,this.keepAliveTime,this.timeUnit,
                new LinkedBlockingQueue<>(64),
                new NamedThreadFactory("BathConsume-"+ this.taskName+"-",true),
                new RejectedPolicyWithReport(this.rejectPolicy));
    }

    public BathConsumeUtil(int poolSize, int maximumPoolSize, int maxRecordSize, String taskName, long maxWaitTime, int keepAliveTime, TimeUnit timeUnit, RejectPolicy rejectPolicy) {
        this.maxRecordSize = maxRecordSize;
        this.keepAliveTime = keepAliveTime;
        this.poolSize = poolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.taskName = taskName;
        this.timeUnit = timeUnit;
        this.maxWaitTime = maxWaitTime;
        this.rejectPolicy = rejectPolicy;
        this.executor = new ThreadPoolExtUtil(this.poolSize,this.maximumPoolSize,this.keepAliveTime,this.timeUnit,
                new LinkedBlockingQueue<>(64),
                new NamedThreadFactory("BathConsume-"+ this.taskName+"-",true),
                new RejectedPolicyWithReport(this.rejectPolicy));
    }

    public void start(){
        check();
        log.info("批处理任务开始,最小线程数:{}========",this.poolSize);
        for (int i = 0; i < poolSize; i++) {
            this.executor.submit(this::run);
        }
    }
    public void stop(){
        stopped = true;
    }
    private void run(){
        LinkedList<T> array = new LinkedList<>();
        long start = 0;
        while (!stopped) {
            T temp;
            int i = 0;
            while ((temp = queue.poll()) != null) {
                array.add(temp); i++;
                if (i >= maxRecordSize)
                    break;
            }

            if (i > 0) {
                start = System.currentTimeMillis();
                try {
                    log.info("task:{}==执行批处理任务，共处理{}条记录",Thread.currentThread().getName(),i);
                    List<T> list = new ArrayList<>();
                    for (int j = 0; j < i; j++) {
                        list.add(array.removeFirst());
                    }
                    runTask(list);
                } catch (Exception e) {
                    log.error("task:{}==执行批处理任务失败,e={},size={}",Thread.currentThread().getName(),e,i);
                }
                long time = System.currentTimeMillis() - start;
                if (time > 1000L)
                    log.warn("task:{}==执行批处理任务耗时较长,共耗时:{}ms,共{}条记录",Thread.currentThread().getName(), time, i);
            }

            if (i < maxRecordSize * 0.75){
                try {
                    if (executor.getActiveCount() > minCorePoolSize){
                        if (i < maxRecordSize * 0.2){
                            toRemove = true;
                            removeTask();
                            log.warn("task:{}==待处理记录数:{},未到达maxRecordSize将阻塞任务maxWaitTime:{}ms,目前活动线程数:{},建议减少线程数或调高maxRecordSize",Thread.currentThread().getName(),queue.size(),maxWaitTime,executor.getActiveCount());
                            log.warn("task:{}==批处理队列未达到最优线程数,将自动缩容.......................",Thread.currentThread().getName());
                        }
                    }
                    Thread.sleep(maxWaitTime);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }else {
                toRemove = false;
            }
            if (queue.size() > maxRecordSize * 10){
                log.warn("task:{}==批处理队列繁忙, 当前活动线程数:{},最大线程数:{},待处理记录数:{},等待队列数：{},建议提高maxRecordSize或者提高核心数poolSize",Thread.currentThread().getName(),executor.getActiveCount(),executor.getMaximumPoolSize(),queue.size(),executor.getQueue().size());
                if (maxWaitTime > 0)
                    maxWaitTime = 0;
                log.warn("task:{}==批处理队列繁忙, 未达到最大线程数,将自动扩容.......................",Thread.currentThread().getName());
                if (executor.getActiveCount() >= maximumPoolSize){
                    for (int j = 0; j < maxRecordSize; j++) {
                        log.warn("task:{}==待处理记录数超出极限,已达到系统最大线程数:{},待处理数量:{},等待队列数：{},将抛弃[{}]条老数据：{}",Thread.currentThread().getName(),executor.getMaximumPoolSize(),queue.size(),executor.getQueue().size(),maxRecordSize,queue.poll());
                    }
                }else {
                    toAdd = true;
                    addTask();
                }
            }else {
                toAdd = false;
            }
        }
    }

    synchronized private void addTask(){
        if (!enableTimer){
            this.countDown(maximumPoolSize-poolSize,30);
        }
        if (!flag.get() && toAdd){
            if (executor.getActiveCount() < maximumPoolSize){
                int newPoolSize = executor.getActiveCount() + 1;
                executor.setCorePoolSize(newPoolSize);
                executor.prestartAllCoreThreads();
                flag.set(true);
                executor.submit(this::run);
            }
            flag.set(true);
        }else {
            log.warn("task:{}----->未执行扩容任务:任务执行标志：{},定时锁定标志：{}，解锁后执行",Thread.currentThread().getName(),toAdd,flag.get());
        }
    }

    synchronized private void removeTask(){
        if (!enableTimer){
            this.countDown(maximumPoolSize-poolSize,60);
        }
        if (!flag.get() && toRemove){
            if (executor.getActiveCount() > minCorePoolSize && executor.getQueue().isEmpty()){
                executor.setCorePoolSize(executor.getActiveCount() - 1);
                executor.prestartAllCoreThreads();
                flag.set(true);
                Thread.currentThread().setDaemon(false);
                Thread.currentThread().interrupt();
                while (!Thread.currentThread().isInterrupted());
            }
            flag.set(true);
        }else {
            log.warn("task:{}<------未执行缩容任务:任务标准：{},定时标志：{}",Thread.currentThread().getName(),toAdd,flag.get());
        }

    }

    /**
     * 动态调整线程倒计时任务
     * @param minute 执行时长 分钟
     * @param interval 间隔时间 秒
     */
    synchronized private void countDown(long minute, long interval){
        if (toAdd != toRemove){
            final Timer timer = new Timer();
            long start = System.currentTimeMillis();
            final long end = start +  minute * 60 * 1000;
            timer.schedule(new TimerTask() {
                @Override
                public void run() {
                    flag.set(false);
                }
            },0,interval * 1000);

            timer.schedule(new TimerTask() {
                @Override
                public void run() {
                    flag.set(false);
                    toAdd = false;
                    toRemove = false;
                    enableTimer = false;
                    timer.cancel();
                }
            },new Date(end));
            enableTimer = true;
        }
    }
    /**
     * 添加数据
     */
    public void add(T t){
        queue.offer(t);
    }

    /**
     * 检查任务是否执行完毕
     */
    public boolean checkTaskOver(){
        if (!this.queue.isEmpty()){
            return false;
        }
        if (this.executor.getActiveCount() > 0){
            log.warn("task:{}==有待处理的任务,任务数:{},将执行线程等待.......",this.taskName,this.executor.getActiveCount());
        }
        log.warn("task:{}==任务执行完毕,共执行任务数:{}",this.taskName,this.executor.getTaskCount());
        return true;
    }

    /**
     * 检查任务是否执行完毕并关闭线程组
     */
    public boolean checkTaskOverAClose(){
        while (!checkTaskOver());
        close();
        return true;
    }

    /**
     * 关闭线程组
     */
    public void close(){
        log.warn("task:{}==任务线程组将会关闭！=====================",this.taskName);
        this.executor.shutdown();
    }

    private void check(){
        if (taskMethod == null)
            throw new RuntimeException("未定义执行批量任务方法,请设置registerMethod方法！");
    }

    private void runTask(List<T> list){
        taskMethod.accept(list);
    }

    /**
     * 注册批处理方法
     * @param taskMethod
     */
    public void registerMethod(Consumer<List<T>> taskMethod){
        this.taskMethod = taskMethod;
    }
}
